package com.qs.javame.thread.pool.mypool;


import java.util.HashSet;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.ReentrantLock;

public class DefaultThreadPool implements ThreadPoolService {

    //线程池中最大线程数
    private static final int MAX_POOL_SIZE = 2;

    //线程池中核心线程池
    private static final int CORE_POOL_SIZE = 1;

    //定义线程池状态
    private static final int RUNNING = -1;
    private static final int SHUTDOWN = 0;
    private static final int STOP = 1;  //shutdownNow()时，直接将线程池置为stop状态
    private static final int TIDYING = 2;
    private static final int TERMINATED = 3;

    //线程池状态默认为为运行状态
    private int runState = RUNNING;

    //线程池中工作的worker数量（工作的线程数）
    private int workerCount;

    //存储工作线程worker
    private HashSet<Worker> workers = new HashSet<>();

    //定义阻塞队列，最多允许存储2个线程任务
    private BlockingQueue<Runnable> workerQueue = new ArrayBlockingQueue<>(2);

    //定义全局锁（workers集合等线程共享资源更新时需要加锁）
    private static final ReentrantLock mainLock = new ReentrantLock();


    //构建默认参数的线程池对象
    public DefaultThreadPool() {

    }

    private void setRunState(int runState) {
        this.runState = runState;
    }

    private boolean isRunning() {
        return runState == RUNNING;
    }

    @Override
    public void execute(Runnable command) {
        if (command == null) {
            throw new NullPointerException("提交的线程任务不能为空！");
        }

        //1、如果当前线程池工作线程数小于核心线程数，则新建worker线程，并执行
        if (workerCount < CORE_POOL_SIZE) {
            if (addWorker(command, true)) {
                return;
            }
        }

        //2、workerCount>corePoolSize，当前线程池是runnable状态，并且workerCount>=corePoolSize，并且任务添加到工作队列成功
        if (isRunning() && workerQueue.offer(command)) {
            //此时，如果工作线程数为0，则新建线程执行workerQueue中存储的任务
            if (workerCount == 0) {
                addWorker(null, true);
            }
        }

        //3、如果workerQueue满了，则需要依赖于线程池设置的最大线程数（在最大线程数以内，允许新建worker线程处理提交的command）
        else if (!addWorker(command, false)) {
            System.out.println("应该执行RejectExecutionHandler....");
        }
    }

    @Override
    public void shutdown() {
        mainLock.lock();
        try {
            //设置线程池的状态为shutdown
            setRunState(SHUTDOWN);
            //中断所有空闲的线程
            interruptIdleWorkers();
        } finally {
            mainLock.unlock();
        }
    }


    private void interruptIdleWorkers() {
        interruptIdleWorkers(false);
    }

    private void interruptIdleWorkers(boolean onlyOne) {
        mainLock.lock();
        try {
            //遍历移除Wokers集合中空闲的worker线程（如何判定线程空闲？）
            for (Worker worker : workers) {
                Thread t = worker.thread;
                //能获取到worker锁，说明线程为空闲线程（也就是线程在执行getTask()方法，或者阻塞在getTask()方法里）
                if (!t.isInterrupted() && worker.tryLock()) {
                    try {
                        t.interrupt();
                    } finally {
                        worker.unlock();
                    }
                }

                if (onlyOne) {
                    break;
                }
            }
        } finally {
            mainLock.unlock();
        }
    }

    @Override
    public boolean addWorker(Runnable firstTask, boolean core) {

        boolean workerStarted = false;

        //当前线程池处于运行状态，则更新workerCount，创建新的worker线程
        //如果当前线程状态为shutdown，则只有满足"runState == SHUTDOWN && !workerQueue.isEmpty() && firstTask == null"条件
        // 才能够创建新的worker线程，执行workerQueue中的任务，初次之外，直接返回false（换言之，处于shutdown状态的线程池不接受新提交的任务，
        // 最多只处理workerQueue中未完成的任务）。
        if (runState >= SHUTDOWN && !(runState == SHUTDOWN && !workerQueue.isEmpty() && firstTask == null)) {
            return false;
        }

        Worker worker = null;
        try {
            //加锁，workerCount属于公共资源
            try {
                mainLock.lock();

                //core决定线程池中创建的线程数最大是为核心线程数还是最大线程数
                if (workerCount >= (core ? CORE_POOL_SIZE : MAX_POOL_SIZE)) {
                    return false;
                }

                //更新线程池中运行中的线程数
                workerCount++;
                System.out.println("-->>workerCount=" + workerCount);

                //将提交的command，封装为worker对象，并启动worker线程
                worker = new Worker(firstTask);
                Thread t = worker.thread;
                if (t != null) {
                    //排除掉线程池处于停止状态
                    if (runState < SHUTDOWN || (runState == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) {
                            throw new IllegalThreadStateException();
                        }

                        //将worker添加到集合中
                        workers.add(worker);

                        //开启worker线程
                        t.start();

                        workerStarted = true;
                    }
                }
            } finally {
                mainLock.unlock();
            }
        } finally {
            //开始失败，移除掉worker
            if (!workerStarted) {
                addWorkerFailed(worker);
            }
        }

        return workerStarted;
    }


    //将添到集合中的worker移除，并更新workerCount
    private void addWorkerFailed(Worker worker) {
        try {
            mainLock.lock();
            if (worker != null) {
                workers.remove(worker);
                workerCount = workerCount - 1;
            }
        } finally {
            mainLock.unlock();
        }
    }

    private void runWorker(Worker worker) {
        //释放锁，线程state=0（默认worker线程初始状态为-1，不允许中断）
        worker.unlock();

        try {
            //获取worker对象中实际的任务（用户提交的command）
            Runnable task = worker.firstTask;

            //getTask()为线程池中空闲的线程从workerQueue中获取任务
            while (task != null || ((task = getTask()) != null)) {
                //获取锁（该锁为不可重入锁，此时线程state=1）
                worker.lock();

                try {
                    //如果线程池已经终止，这里需要中断线程
                    if (runState >= STOP && !Thread.currentThread().isInterrupted()) {
                        Thread.currentThread().interrupt();
                    }

                    Throwable thrown = null;
                    try {
                        //执行任务
                        task.run();
                    } catch (Exception e) {
                        thrown = e;
                        throw e;
                    } finally {
                        afterExecute(task, thrown);
                    }
                } finally {
                    //重置task
                    task = null;

                    //释放锁（此时线程state=0）
                    worker.unlock();
                }
            }
        } finally {
            processWorkerExit(worker);
        }
    }

    private void processWorkerExit(Worker worker) {
        mainLock.lock();
        try {
            //移除已经完成的任务
            workers.remove(worker);
        } finally {
            mainLock.unlock();
        }
    }

    private void afterExecute(Runnable task, Throwable thrown) {
        //子类重写
    }

    //线程池中的线程，在执行完firstTask之后，总是会进入该方法，不断循环或者阻塞等待获取阻塞队列中的任务，获取到任务返回，获取不到任务则
    //一直循环重复获取或者阻塞等待，除非该线程被中断停止。
    private Runnable getTask() {
        for (; ; ) {
            if (runState >= SHUTDOWN && workerQueue.isEmpty()) {
                return null;
            }

            try {
                return workerQueue.take();
            } catch (InterruptedException e) {
                System.out.println(Thread.currentThread().getName() + "-被中断了~");
                e.printStackTrace();
            }
        }
    }


    class Worker extends AbstractQueuedSynchronizer implements Runnable {

        private volatile boolean isIdle = false;

        Runnable firstTask;

        private Thread thread;


        public Worker(Runnable firstTask) {
            this.firstTask = firstTask;
            this.thread = ThreadFactory.getInstance().newThread(this);
            //设置当前线程状态为-1
            setState(-1);
        }

        //worker线程的线程任务
        @Override
        public void run() {
            runWorker(this);
        }


        protected boolean isHeldExclusively() {
            return getState() != 0;
        }

        protected boolean tryAcquire(int unused) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        protected boolean tryRelease(int unused) {
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        public void lock() {
            acquire(1);
        }

        public boolean tryLock() {
            return tryAcquire(1);
        }

        public void unlock() {
            release(1);
        }

        public boolean isLocked() {
            return isHeldExclusively();
        }

        //只有获得锁，并且没有中断的线程，才允许中断
        void interruptIfStarted() {
            Thread t;
            if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                }
            }
        }
    }


    //线程工厂类
    static class ThreadFactory {

        //线程名计数前缀
        private AtomicInteger count = new AtomicInteger(0);

        //饿汉式单例
        private static ThreadFactory threadFactory = new ThreadFactory();

        private ThreadFactory() {

        }

        //单例
        public static ThreadFactory getInstance() {
            return threadFactory;
        }

        public Thread newThread(Runnable worker) {
            return new Thread(worker, "my-pool-" + count.getAndIncrement() + "-thread");
        }
    }
}


class TestClient {
    public static void main(String[] args) {
        DefaultThreadPool defaultThreadPool = new DefaultThreadPool();
        defaultThreadPool.execute(new Command(0));
        defaultThreadPool.execute(new Command(1));
        defaultThreadPool.execute(new Command(2));
        defaultThreadPool.execute(new Command(3));
        defaultThreadPool.execute(new Command(4));

        defaultThreadPool.shutdown();
    }
}


class TestClient2 {
    public static void main(String[] args) {
        long startTime = System.currentTimeMillis();

        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 2, 0,
                TimeUnit.SECONDS, new ArrayBlockingQueue<>(2));
        threadPoolExecutor.execute(new Command(0));
        threadPoolExecutor.execute(new Command(1));
        threadPoolExecutor.execute(new Command(2));
        threadPoolExecutor.execute(new Command(3));
//        threadPoolExecutor.execute(new Command(4));

        boolean flag = false;

        try {
            threadPoolExecutor.shutdown();
            while (!flag) {
                flag = threadPoolExecutor.awaitTermination(10, TimeUnit.SECONDS);
            }

            System.out.println("线程池停止成功，总耗时：" + (System.currentTimeMillis() - startTime) / 1000 + "s");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
}


class Command implements Runnable {

    private int i;

    public Command(int i) {
        this.i = i;
    }

    @Override
    public void run() {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println(Thread.currentThread().getName() + "-执行command-" + i + "-任务~");
    }
}